package com.bailiny.hdfs.demo1;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.testng.annotations.Test;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;

/**
 * @author Yang Xu
 * @date 2020/1/13 9:04
 * @description:
 */
public class HdfsOperateStudy {

    private String hdfsUrl = "hdfs://192.168.52.100:8020";

    /**
     * 使用URL的方式访问数据
     */
    @Test
    public void test() {
        // 1.注册URL驱动
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

        try {
            InputStream inputStream = new URL("hdfs://192.168.52.100:8020/test/input/install.log").openStream();
            FileOutputStream outputStream = new FileOutputStream(new File("D:\\tmp\\test.txt"));
            // 使用输入流读取文件
            // 使用输出流将文件写到哪去
            // 通过new url的方式打开一个文件流
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(outputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过fileSystem获取分布式文件系统的几种方式
     */
    @Test
    public void getFileSystem1() throws IOException {
        //如果configuration 不做任何配置，获取到的是本地文件系统
        Configuration configuration = new Configuration();
        //覆盖我们的hdfs的配置，得到我们的分布式文件系统
        configuration.set("fs.defaultFS", hdfsUrl);
        FileSystem fileSystem = FileSystem.get(configuration);
        System.out.println(fileSystem.toString());
    }

    /**
     * 获取hdfs的第二种方式
     */
    @Test
    public void getHdfs2() throws URISyntaxException, IOException {
        //使用两个参数来获取hdfs文件系统
        //第一个参数是一个URI，定义了我们使用hdfs://这种方式来访问，就是分布式文件系统
        //
        FileSystem fileSystem = FileSystem.get(new URI(hdfsUrl), new Configuration());
        System.out.println(fileSystem.toString());
    }


    /**
     * 获取hdfs分布式文件系统的第三种方式
     */
    @Test
    public void getHdfs3() throws IOException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", hdfsUrl);
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        System.out.println(fileSystem.toString());
    }

    /**
     * 获取hdfs分布式文件系统的第四种方式  通过newInstance传递两个参数
     */


    /**
     * 递归遍历查询所有的文件夹和文件
     *
     * @throws Exception
     */
    @Test
    public void listFile() throws Exception {
        FileSystem fileSystem = FileSystem.newInstance(new URI(hdfsUrl), new Configuration());
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isDirectory()) {
                Path path = fileStatus.getPath();
                listAllFiles(fileSystem, path);
            } else {
                System.out.println("文件路径为" + fileStatus.getPath().toString());
            }

        }
    }

    public void listAllFiles(FileSystem fileSystem, Path path) throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(path);
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isDirectory()) {
                System.out.println("文件夹路径为" + fileStatus.getPath().getName());
                listAllFiles(fileSystem, fileStatus.getPath());
            } else {
                Path path1 = fileStatus.getPath();
                System.out.println("文件路径为" + path1);
            }
        }
    }

    /**
     * 拷贝文件到本地
     *
     * @throws Exception
     */
    @Test
    public void getFileToLocal() throws Exception {
        FileSystem fileSystem = FileSystem.newInstance(new URI(hdfsUrl), new Configuration());
        FSDataInputStream open = fileSystem.open(new Path("/bigfile.xml"));
        FileOutputStream outputStream = new FileOutputStream(new File("D:/tmp/bigfile.xml"));

        IOUtils.copy(open, outputStream);
        IOUtils.closeQuietly(open);
        IOUtils.closeQuietly(outputStream);
        fileSystem.close();
        System.out.println("拷贝成功");
    }

    /**
     * 创建文件夹
     *
     * @throws Exception
     */
    @Test
    public void mkdir() throws Exception {
        FileSystem fileSystem = FileSystem.newInstance(new URI(hdfsUrl), new Configuration());
        fileSystem.mkdirs(new Path("/hello/mydir/test"));
        fileSystem.close();
    }

    /**
     * 文件上传
     *
     * @throws Exception
     */
    @Test
    public void putData() throws Exception {
        FileSystem fileSystem = FileSystem.newInstance(new URI(hdfsUrl), new Configuration());
        fileSystem.copyFromLocalFile(new Path("D:/tmp/install.log"), new Path("/hello/mydir/test/install.log"));
        fileSystem.close();
    }

    /**
     * 下载文件，修改hdfs-site.xml permission为true后需要伪造用户
     *
     * @throws Exception
     */
    @Test
    public void getConfig() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI(hdfsUrl), new Configuration(), "root");
        fileSystem.copyToLocalFile(new Path("/config/core-site.xml"), new Path("file:///d:/tmp/core-site.xml"));
        fileSystem.close();
    }

    /**
     * 将本地多个文件上传到hdfs，合并成为一个大的文件
     *
     * @throws Exception
     */
    @Test
    public void mergeFile() throws Exception {
        //获取分布式文件系统
        FileSystem fileSystem = FileSystem.get(new URI(hdfsUrl), new Configuration(), "root");
        FSDataOutputStream outputStream = fileSystem.create(new Path("/bigfile.xml"));
        //获取本地文件系统
        LocalFileSystem local = FileSystem.getLocal(new Configuration());
        //通过本地文件系统获取文件列表，为一个集合
        FileStatus[] fileStatuses = local.listStatus(new Path("file:///D:\\DD_big_data\\大数据新版\\18年12月版大数据课件\\04 -hdfs hdfs资料\\3、大数据离线第三天\\上传小文件合并"));
        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream = local.open(fileStatus.getPath());
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        IOUtils.closeQuietly(outputStream);
        local.close();
        fileSystem.close();
    }

}
